package com.hung.trade.common.rocketmq;

import com.hung.trade.common.exception.TradeException;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * @author hungki
 * @createTime 2018-10-08 上午 8:58
 */
public class RocketMQConsumer {
    public static final Logger logger = LoggerFactory.getLogger(RocketMQConsumer.class);
    private String groupName;
    private String topic;
    private String tags="*";
    private String nameservAddr;
    private int consumeThreadMin = 20;
    private int consumeThreadMax =64;

    public void setGroupName(String groupName) {
        this.groupName = groupName;
    }

    public void setTopic(String topic) {
        this.topic = topic;
    }

    public void setTags(String tags) {
        this.tags = tags;
    }

    public void setNameservAddr(String nameservAddr) {
        this.nameservAddr = nameservAddr;
    }

    public void setConsumeThreadMax(int consumeThreadMax) {
        this.consumeThreadMax = consumeThreadMax;
    }

    public void setConsumeThreadMin(int consumeThreadMin) {
        this.consumeThreadMin = consumeThreadMin;
    }

    public void setiMessageProcesser(IMessageProcesser iMessageProcesser) {
        this.iMessageProcesser = iMessageProcesser;
    }

    private IMessageProcesser iMessageProcesser;

    public void init() throws TradeException {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(groupName);
        consumer.setNamesrvAddr(nameservAddr);
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        try {
            consumer.subscribe(topic,tags);
            TradeMessageListener tradeMessageListener = new TradeMessageListener(iMessageProcesser);
            consumer.setMessageListener(tradeMessageListener);
            consumer.start();
            logger.info(String.format("consumer is start [groupName = [%s],topic = [%s],tags = [%s]]\n",groupName,topic,tags));
        } catch (MQClientException e) {
            logger.error(String.format("subscribe[topci = [%s]] error!",topic),e);
            throw new TradeException(e);
        }
    }
}
